package com.guchenbo.flink;

import com.guchenbo.datagen.common.datatypes.TaxiRide;
import com.guchenbo.datagen.common.sources.TaxiRideGenerator;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author guchenbo
 * @date 2024/6/10
 */
public class TaxiStreamDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        DataStream<TaxiRide> stream = env.addSource(new TaxiRideGenerator());
        SingleOutputStreamOperator<Long> streamOperator = stream.map(taxi -> taxi.taxiId);
        KeyedStream<Long, Long> keyedStream = streamOperator.keyBy(k -> k);
        keyedStream.print();
        System.out.println(env.getParallelism());
                env.execute();
    }
}
